package cn.gmy.progect1;

import java.io.IOException;
import java.io.InputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class ExcelInputFormat extends FileInputFormat<LongWritable, Text>{
	@Override
	public RecordReader<LongWritable, Text> createRecordReader(InputSplit inputSplit, TaskAttemptContext context)
			throws IOException, InterruptedException {		
		return new ExcelRecordReader();
	}

	//创建ExcelRecordReader类
	public static class ExcelRecordReader extends RecordReader<LongWritable, Text>{
		private String[] stringAll=null;
		private InputStream inputStream;
		private LongWritable key=new LongWritable(0);
		private Text value = new Text();
		
		@Override
		public void initialize(InputSplit inputSplit, TaskAttemptContext context) throws IOException, InterruptedException {
			Configuration configuration=context.getConfiguration();
			FileSystem fSystem=FileSystem.get(configuration);
			FileSplit fileSplit = (FileSplit) inputSplit;
			Path path = fileSplit.getPath();			
			String fileName = path.getName();			
			inputStream = fSystem.open(path);
			this.stringAll=ExcelParser.parseExcelData(inputStream,fileName);
		}

		@Override
		public void close() throws IOException {
			if (inputStream!=null) {
				inputStream.close();
			}
		}
		
		@Override
		public boolean nextKeyValue() throws IOException, InterruptedException {
			int rowCount = (int) (key.get());	
			if (rowCount<stringAll.length) {
				if (stringAll[rowCount]!=null) {
					key.set(rowCount+1);
					value.set(stringAll[rowCount]);
					return true;
				}
			}		
			return false;
		}

		@Override
		public LongWritable getCurrentKey() throws IOException, InterruptedException {
			return key;
		}

		@Override
		public Text getCurrentValue() throws IOException, InterruptedException {
			return value;
		}

		@Override
		public float getProgress() throws IOException, InterruptedException {
			return 0;
		}
	}	
}
